package com.hr.reactive.flux.assignment;

import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Objects;

/**
 * @author huran
 */
public class FileReaderServiceImpl implements FileReaderService {
    @Override
    public Flux<String> read(Path path) {
        return Flux.generate(() -> openFile(path),this::readFile,this::closeFile);
    }

    private  BufferedReader openFile(Path path) throws IOException {
        return Files.newBufferedReader(path);

    }
    private void closeFile(BufferedReader reader){
        try {
            reader.close();
            System.out.println("close file");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    private  BufferedReader readFile(BufferedReader reader, SynchronousSink<String> synchronousSink) {
        try {
            String s = reader.readLine();
            if (Objects.isNull(s)) {
                synchronousSink.complete();
            } else {
                synchronousSink.next(s);
            }
        } catch (Exception e) {
            synchronousSink.error(e);
        }
        return reader;
    }
}
